package com.test.lock;

import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;


/**
 * 信号量实现的多线程操作
 */
@Slf4j
public class SemaphoreMain {

    /**
     * 信号量为2的公平锁:限流，每次并行处理2条数据
     */
    private static Semaphore semaphore = new Semaphore(3, true);
    private static ConcurrentLinkedQueue<Integer> container = new ConcurrentLinkedQueue();
    private static volatile List<Integer> list = new ArrayList();
    private static ExecutorService poolService = Executors.newFixedThreadPool(8);

    // 需要处理的任务坐标
    private static volatile Integer index = 0;
    private static AtomicInteger atomicindex = new AtomicInteger(0);
    private static Integer total = 50;
    // private static CountDownLatch countDownLatch = new CountDownLatch(3);
    private static CyclicBarrier cyclicBarrier = new CyclicBarrier(3);


    public static void main(String[] args) {
         consumeSafe();
        // consumePollSafe();
        // consumeArrayListUnsafe();
//        consumeListUnSafe();
        produce();

        try {
            TimeUnit.MILLISECONDS.sleep(500L);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        log.info("end");
        poolService.shutdownNow();
        System.exit(1);
    }

    /**
     * 往container里添加任务
     */
    public static void produce() {
        Thread producer = new Thread(() -> {
            while (true) {
                if (container.size() >= total || list.size() >= total) {
                    /*if (cyclicBarrier.getParties() == total) {
                        cyclicBarrier.reset();
                    }*/
                    log.info("生产者暂停500毫秒");
                    try {
                        TimeUnit.MILLISECONDS.sleep(500L);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } finally {
                        // 循环生产任务
                        // produce();
                    }
                }
                Integer i = new Random().nextInt(100);
                log.info("生产：{}", i);
                container.offer(i);
//                list.add(i);
                // atomicindex.incrementAndGet(); 此时消费者需要减少
                /*try {
                    TimeUnit.MILLISECONDS.sleep(5L);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    // 循环生产任务
                    // produce();
                }*/
            }
        });
        // 2个生产者生产任务
        poolService.execute(producer);
        poolService.execute(producer);
        log.info("生产者已启动");
    }

    /**
     * 从container里取任务
     */
    public static void consumeSafe() {
        Thread consumer = new Thread(() -> {
            while (true) {
                if (container.isEmpty()) {
                    log.info("没有任务，等待中...");
                    try {
                        TimeUnit.MILLISECONDS.sleep(100L);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    continue;
                }
                log.info("待处理任务：{}", container);
                Integer i = container.poll();
                if (i == null) {
                    log.info("待处理任务是null，等待中...");
                    try {
                        TimeUnit.MILLISECONDS.sleep(100L);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    continue;
                }
                try {
                    if (semaphore.availablePermits() == 0) {
                        log.info(i + "没有可用信号量，等待中...");
                    }
                    semaphore.acquire();
                    log.info("开始执行 " + i);
                    // 模拟程序执行耗时
                    Thread.sleep(10L);
                } catch (Throwable e) {
                    e.printStackTrace();
                } finally {
                    semaphore.release();
                    log.info("执行完 " + i);
                    // 循环消费任务 consume();
                }
            }
        });
        // 3个消费者处理任务
        poolService.execute(consumer);
        poolService.execute(consumer);
        poolService.execute(consumer);
        /* consumer.start();
        consume();*/
        log.info("消费者已启动");
    }

    /**
     * 从List里取任务
     * 使用AtomicInteger来保证每次处理的任务坐标不重复，不遗漏，破解线程安全问题
     */
    public static void consumeListUnSafe() {
        Thread consumer = new Thread(() -> {
            while (true) {
                /*try {
                    Thread.sleep(1L);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }*/
//                if (index >= 10) {
//                    list.clear();
//                }

                Integer indexNum = atomicindex.getAndIncrement();
                log.info("处理第{}个任务", indexNum);
                if (indexNum < 0) {
                    atomicindex.set(0);
                    /*try {
                        // 休眠10毫秒，防止直接清空list导致其他并行处理的线程获取不到数据
                        TimeUnit.MILLISECONDS.sleep(10L);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }*/
                    continue;
                }
                if (indexNum >= total) {
                    atomicindex.set(0);
                    list.clear();
                    container.clear();
                    // countDownLatch.countDown();
                    try {
                        log.info("等待生成任务...");
                        cyclicBarrier.await();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } catch (BrokenBarrierException e) {
                        e.printStackTrace();
                    }
                    continue;
                }
                /*if (countDownLatch.getCount() < 1) {
                    list.clear();
                    try {
                        // 休眠10毫秒，防止直接清空list导致其他并行处理的线程获取不到数据
                        TimeUnit.MILLISECONDS.sleep(10L);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }*/
                if (list.isEmpty()) {
                    log.info("没有任务，等待中...");
                    atomicindex.decrementAndGet();
                    /*try {
                        TimeUnit.MILLISECONDS.sleep(200L);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }*/
                    continue;
                }
                try {
                    /*if (semaphore.availablePermits() == 0) {
                        log.info(Thread.currentThread().getName() + " 没有可用信号量，等待中...");
                    }
                    semaphore.acquire();*/
                    System.out.println(list.size());
                    System.out.println(indexNum);

                    if (list.size() >= indexNum) {
                        log.info("没有任务，等待中...");
                        continue;
                    }
                    Integer i = list.get(indexNum); // list.remove(0); // list.isEmpty() ? null : list.remove(0);
                    if (i == null) {
                        log.info("待处理任务是null，等待中...");
                        /*try {
                            TimeUnit.MILLISECONDS.sleep(100L);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }*/
                        continue;
                    }
                    log.info("开始执行 " + i);
                    // ++index;
                    // 模拟程序执行耗时
                    /*Thread.sleep(50L);*/
                } catch (Throwable e) {
                    e.printStackTrace();
                } finally {
                    /*semaphore.release();*/
                    // 循环消费任务 consume();
                }
            }
        });
        // 3个消费者处理任务
        poolService.execute(consumer);
        poolService.execute(consumer);
        poolService.execute(consumer);
        /* consumer.start();
        consume();*/
        log.info("消费者已启动");
    }

    /**
     * 从container里取任务
     */
    public static void consumePollSafe() {
        Thread myThread = new MyThread();
        // 3个消费者处理任务
        poolService.execute(myThread);
        poolService.execute(myThread);
        poolService.execute(myThread);
        log.info("消费者已启动");
    }

    static class MyThread extends Thread {
        public void handle() {
            while (true) {
                if (container.isEmpty()) {
                    log.info("没有任务，等待中...");
                    try {
                        TimeUnit.MILLISECONDS.sleep(200L);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    continue;
                }
                try {
                    Thread.sleep(0L);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                log.info("待处理任务：{}", container);
                Integer i = container.poll();
                if (i == null) {
                    log.info("待处理任务是null，等待中...");
                    try {
                        TimeUnit.MILLISECONDS.sleep(200L);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    continue;
                }
                log.info("开始执行 " + i);
            }
        }

        @Override
        public void run() {
            handle();
        }
    }

    /**
     * 从container里取任务
     */
    public static void consumeArrayListUnsafe() {
        Thread myThread = new MyThreadOfArrayList();
        // 3个消费者处理任务
        poolService.execute(myThread);
        poolService.execute(myThread);
        poolService.execute(myThread);
        log.info("消费者已启动");
    }

    /**
     * 线程不安全
     */
    static class MyThreadOfArrayList extends Thread {

        public void handle() {
            if (list.isEmpty()) {
                log.info("没有任务，等待中...");
                try {
                    TimeUnit.MILLISECONDS.sleep(200L);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                handle();
            }
            // log.info("待处理任务：{}", list);
            Integer i = list.remove(0);
            if (i == null) {
                log.info("待处理任务是null，等待中...");
                try {
                    TimeUnit.MILLISECONDS.sleep(200L);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                handle();
            }
            log.info("开始执行 " + i);
            handle();
        }

        @Override
        public void run() {
            handle();
        }
        /* 存在重复处理和丢失数据的问题
        2023-08-10 14:50:00.849 [pool-2-thread-5] INFO  com.test.lock.SemaphoreMain:48 - 生产者暂停500毫秒
        2023-08-10 14:50:01.001 [pool-2-thread-3] INFO  com.test.lock.SemaphoreMain:212 - 开始执行 34
        2023-08-10 14:50:01.001 [pool-2-thread-2] INFO  com.test.lock.SemaphoreMain:212 - 开始执行 34 重复处理
        2023-08-10 14:50:01.001 [pool-2-thread-2] INFO  com.test.lock.SemaphoreMain:212 - 开始执行 33
        2023-08-10 14:50:01.001 [pool-2-thread-2] INFO  com.test.lock.SemaphoreMain:212 - 开始执行 59
        2023-08-10 14:50:01.002 [pool-2-thread-2] INFO  com.test.lock.SemaphoreMain:212 - 开始执行 59 重复处理
        2023-08-10 14:50:01.002 [pool-2-thread-2] INFO  com.test.lock.SemaphoreMain:212 - 开始执行 80
        2023-08-10 14:50:01.002 [pool-2-thread-2] INFO  com.test.lock.SemaphoreMain:212 - 开始执行 21
        2023-08-10 14:50:01.002 [pool-2-thread-2] INFO  com.test.lock.SemaphoreMain:212 - 开始执行 64
        2023-08-10 14:50:01.002 [pool-2-thread-2] INFO  com.test.lock.SemaphoreMain:212 - 开始执行 11
        2023-08-10 14:50:01.002 [pool-2-thread-2] INFO  com.test.lock.SemaphoreMain:212 - 开始执行 13
         */

    }
}
